GoogleCloud WorkflowsのGoogle Sheets APIコネクタでGoogleスプレッドシートのデータをBigQueryに取り込んでみる

kobayashi.m
2024.08.31
 はじめにデータアナリティクス事業本部のkobayashiです。
前回はGoogleCloudのWorkflowsのBigQuery Data TransferコネクタがGAされていたのでGCSにあるCSVファイルをBigQueryにロードする処理をこのコネクタを使って実装してみましたが、今回も同じタイミングでGAされていたGoogle Sheets APIコネクタを試してみたのでまとめます。
https://dev.classmethod.jp/articles/googlecloud-workflows-bigquery-data-transfer/
Workflows release notes  |  Google Cloud
Google Sheets API Connector Overview  |  Workflows  |  Google Cloud
 Google Sheets API Connector 対象のGoogleスプレッドシートとBigQueryテーブル対象となるGoogle スプレッドシートとBigBigQueryのデータセットとテーブルは下記のものを使います。

スプレッドシートの中身は前回使ったcsvファイルをスプレッドシートにしたもので以下の天候状況になります。

読み込む際にスプレッドシートIDとスプレッドシートの共有が必要なので以降に説明をいたします。

BigQueryのテーブルのスキーマ情報は前回使ったテーブルと同じテーブルを使用します。
$ bq show --schema --format=prettyjson data_set_weather.jp_weather
[
  {
    "mode": "NULLABLE",
    "name": "date",
    "type": "DATE"
  },
  {
    "mode": "NULLABLE",
    "name": "month",
    "type": "INTEGER"
  },
  {
    "mode": "NULLABLE",
    "name": "city_name",
    "type": "STRING"
  },
  {
    "mode": "NULLABLE",
    "name": "weather_condition",
    "type": "STRING"
  },
  {
    "mode": "NULLABLE",
    "name": "temperature",
    "type": "NUMERIC"
  },
  {
    "mode": "NULLABLE",
    "name": "humidity",
    "type": "NUMERIC"
  },
  {
    "mode": "NULLABLE",
    "name": "sunshine_duration",
    "type": "NUMERIC"
  },
  {
    "mode": "NULLABLE",
    "name": "cloud_cover",
    "type": "NUMERIC"
  }
]
 ワークフロー定義を記述ワークフローをWorkflows構文に従ってyamlにて記述します。WorkflowsのGoogle Sheets APIコネクタに関する公式ドキュメントがありますのでこのサンプルを参考に記述します。
Google Sheets API Connector Overview  |  Workflows  |  Google Cloud
 jp-weather-sheets2bq.yml
main:
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - destination_dataset: "data_set_weather" # ターゲットとなるBigQueryのデータセット
          - destination_table: "jp_weather" # ターゲットとなるBigQueryのテーブル
          - location: "asia-northeast1"
          - sheetId: "{スプレッドシートID}" #
          - getRange: "Sheet1!A2:H50"  # 読み込む範囲のA1表記またはR1C1表記
          - insert_data: ""
    - get_from_sheet:
        call: googleapis.sheets.v4.spreadsheets.values.get
        args:
          range: ${getRange}
          spreadsheetId: ${sheetId}
          dateTimeRenderOption:
        result: getResult

    - process_result:
        for:
          value: row
          in: ${getResult.values}
          steps:
            - process_each_row:
                assign:
                  - date: ${row[0]}
                  - month: ${row[1]}
                  - city_name: ${row[2]}
                  - weather_condition: ${row[3]}
                  - temperature: ${row[4]}
                  - humidity: ${row[5]}
                  - sunshine_duration: ${row[6]}
                  - cloud_cover: ${row[7]}
                  - row:  ${"('"+date+"', "+month+", '"+city_name+"', '"+weather_condition+"', "+temperature+", "+humidity+", "+sunshine_duration+", "+cloud_cover +")"}
                  - sep: ${if(len(insert_data) > 0, ",", "")}
                  - insert_data: ${insert_data + sep + row}
    - construct_query:
        assign:
          - query: ${"INSERT INTO `" + project_id + "." + destination_dataset + "." + destination_table + "` (date, month, city_name, weather_condition, temperature, humidity, sunshine_duration, cloud_cover) VALUES " + insert_data}

    - insert_data:
        call: googleapis.bigquery.v2.jobs.query
        args:
          projectId: ${project_id}
          body:
            query: ${query}
            useLegacySql: false
        result: queryResult
    - the_end:
        return: ${queryResult}
処理の中身を解説します。
大枠は公式のサンプル(Google Sheets API Connector Overview  |  Workflows  |  Google Cloud  )のgoogleapis.sheets.v4.spreadsheets.values.getメソッドを使用しスプレッドシートからデータを抽出し、取得したデータを加工後BigQueryコネクタでInsertするQueryを実行しています。
ステップを詳しく見ていきます。
initステップではBigQuerのテーブル情報とソースとなるスプレッドシートの情報を設定しています。注意したいのがsheetIdとgetRangeです。
sheetIdはGoogleスプレッドシートを開くと表示されるURLがhttps://docs.google.com/spreadsheets/d/{スプレッドシートID}/edit?gid=0#gid=0 となるのでこのURLから取得します。またgetRangeですがA1表記かR1C1表記となります（Google Sheets API の概要  |  Google for Developers ）。

get_from_sheetステップではGoogle Sheets APIコネクタを使ってinitステップで指定したシートからgetRangeの範囲でデータを取得します。取得したデータの形式はValueRange型（Types overview  |  Workflows  |  Google Cloud ）となりますので、使いたいスプレッドシートのデータはvaluesにList型入ってきます。
process_resultステップではget_from_sheetステップで取得したデータがList型なのでこれを加工してInsert文のValue句にいれる文字列を整形しています。整形にはfor:in ループを使ってValueRange.valuesから1行ごとにデータを取り出し、文字列結合してValue句で指定するデータの文字列を作成しています。
construct_queryテップではBiqQueryで実行するSQLを組み立てています。Insert文でテーブルとprocess_resultステップで作成したデータの文字列をValue句に指定します。
insert_dataステップではBigQueryコネクタでconstruct_queryテップで作成したInsert文を実行します。
 ワークフローの作成と実行それでは記述したワークフロー定義をWorkflowsに登録し、ワークフローを実行してみます。

ワークフローを実行するためにはサービスアカウントが必要になりますがこちらは前回使用したものと同じサービスアカウントを使ってワークフローを作成します。
gcloud workflows deploy jp-weather-sheets2bq --source=jp-weather-sheets2bq.yml --service-account jp-weather-2@{プロジェクト名}.iam.gserviceaccount.com --location asia-northeast1
これでワークフローの作成は完成です。 Cloudコンソールから確認すると以下のようにフローが図になって表現されています。

これでワークフローは作成されましたが、今回はGoogleスプレッドシートを操作するためAPIを有効化するのと対象のスプレッドシートにサービスアカウントからアクセスできるようにする必要があります。
Google ドライブ、Google スプレッドシートAPIの有効化をする
$ gcloud services enable drive.googleapis.com sheets.googleapis.com
スプレッドシートにサービスアカウントからアクセスできるように共有設定を行う

Google スプレッドシートの右上の共有を押下して、gcloud workflows deployで使用したサービスアカウントのメールアドレス（今回だとjp-weather-2@{プロジェクト名}.iam.gserviceaccount.com）を登録する。
これらの設定を行わないとワークフローの実行時にエラーになるので必ず設定を行います。

では作成したワークフローを実行します。
$ gcloud workflows run jp-weather-sheets2bq --location asia-northeast1
Waiting for execution [89b765cc-eea4-45df-b107-09fd1c222abe] to complete...done.
...
startTime: '2024-08-30T11:33:40.494290Z'
state: SUCCEEDED
status:
  currentSteps:
  - routine: main
    step: the_end
workflowRevisionId: 000041-429
yamlの記述やサービスアカウントの権限に問題がなければ特に問題なくワークフローの実行が終わります。

処理が終わりBigQueryで確認するとデータが登録されていることがわかります。
$ bq query "select * from data_set_weather.jp_weather limit 10" 
+------------+-------+-----------+-------------------+-------------+----------+-------------------+-------------+
|    date    | month | city_name | weather_condition | temperature | humidity | sunshine_duration | cloud_cover |
+------------+-------+-----------+-------------------+-------------+----------+-------------------+-------------+
2024-08-10   |     1 | 東京       | 晴                |        27.8 |       79 |               0.0 |           1 |
2024-08-10   |     2 | 東京       | 晴                |        27.8 |       80 |               0.0 |           1 |
2024-08-10   |     3 | 東京       | 晴                |        27.5 |       80 |               0.0 |           1 |
2024-08-10   |     4 | 東京       | 晴                |        27.5 |       82 |               0.0 |           1 |
2024-08-10   |     5 | 東京       | 晴                |        27.5 |       80 |               0.0 |           0 |
2024-08-10   |     6 | 東京       | 晴                |        27.4 |       76 |               0.0 |           0 |
2024-08-10   |     7 | 東京       | 晴                |        28.3 |       72 |               0.4 |           0 |
2024-08-10   |     8 | 東京       | 晴                |        29.4 |       67 |               0.9 |           0 |
2024-08-10   |     9 | 東京       | 晴                |        31.2 |       62 |               1.0 |           0 |
2024-08-10   |    10 | 東京       | 晴                |        31.0 |       59 |               0.5 |           0 |
+------------+-------+-----------+-------------------+-------------+----------+-------------------+-------------+
 まとめGoogleCloudのWorkflowsでGoogle Sheets APIコネクタを使ってGoogle スプレッドシートのデータをBigQueryに読み込んでみました。Google Sheets APIコネクタを使うことでWorkflowsでもスプレッドシートの操作を行えるので色々応用が効きそうです。
最後まで読んで頂いてありがとうございました。